KAFKA-7140: Remove deprecated poll usages by viktorsomogyi · Pull Request #5319 · apache/kafka

viktorsomogyi · 2018-07-02T12:58:39Z

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

ijuma · 2018-07-02T20:03:12Z

Thanks for the PR. Maybe we can update all non-test usages of the deprecated method at once?

viktorsomogyi · 2018-07-09T13:59:41Z

@ijuma thanks for the suggestion, I've updated the PR.

viktorsomogyi · 2018-07-09T14:02:06Z

@mjsax I think you authored the change where this line was introduced. Do you know what's the reasoning for poll(1)? Did it just used the fact that it updated the metadata or were there any other reasons? I'm not sure that my change is correct here and I don't have the context.

Yes, we do this call to update the metadata. The change seems reasonable to me.

Are you sure about this? poll(1) would previously have guaranteed that the consumer joins the group. That is no longer the case with the new poll, so the call to assignment below is likely to return an empty set.

Because StreamsResetter must be the only member in the group (we check on startup that the group is empty), we can check if the assignment contains all offsets? Or we could switch to manual assignment?

Yeah, I think manual assignment is a better option here. For one thing, it guarantees that there won't be another member in the group when you receive the assignment. It also gets around the awkward use of poll() since you do not actually want to fetch data. The basic approach would be to use the partitionsFor API to find the partitions for the subscribed topics and then assign them directly with assign.

Ok, I'll try to refactor my code this way.

viktorsomogyi · 2018-07-09T14:05:59Z

Also created a jira for this as it I think it now exceeds the "minor" category :)

hachikuji · 2018-07-19T21:08:58Z

@viktorsomogyi Thanks for the patch. Note that there is a semantic difference between the old and new poll APIs. The old API blocked if a rebalance was in progress, but the new one does not. You may need to review each usage individually to be sure that the new semantic is still appropriate.

viktorsomogyi · 2018-08-08T13:49:36Z

retest this please

viktorsomogyi · 2018-08-08T13:53:33Z

@hachikuji thanks for the note. Yes, I've been trying to keep in mind that poll(0) differs in this case. In some cases we could just use to the position call or poll(0) just because we make sure to have an initial value to an iterator, etc.
The only case I found so far where I'm not sure is in StreamsResetter.

hachikuji

Apologies for the delay. Left a couple comments.

hachikuji · 2018-08-08T16:22:38Z

Maybe we can just initialize this to an empty iterator? The call seems intended to join the group, but that will no longer happen when the call returns.

Were you thinking about something like this?
var recordIter = Collections.emptyList[ConsumerRecord[Array[Byte], Array[Byte]]]().iterator()

hachikuji · 2018-08-08T16:27:49Z

As far as I can tell, the assignment will be empty here. I don't think the old code worked either as intended because it did the seekToEnd prior to the call to poll(). For an end-to-end latency test, I actually wonder if we should just use manual assignment and take the group management out of the picture?

Will change to manual assignment, something like this (which seems to work).

val topicPartitions = consumer.listTopics().get(topic).asScala .map(pi => new TopicPartition(pi.topic(), pi.partition())) .to[List].asJava consumer.assign(topicPartitions) consumer.seekToEnd(topicPartitions) consumer.assignment().asScala.foreach(consumer.position)

viktorsomogyi · 2018-08-09T15:01:13Z

So here it seems the old code assumed that there is a topic. Now I added a check that there must be a topic beforehand. the producer would anyway create it according to the default settings but I think that most of the latency tests executed by users won't depend on the default settings and they rather want to create topics manually according to their likings.

I guess another approach might be to produce a dummy record to the topic first before assigning the partitions to the consumer and starting the test. That would let us retain the existing behavior.

…ndLatency

viktorsomogyi · 2018-08-09T16:48:19Z

        try (final KafkaConsumer<byte[], byte[]> client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
-            client.subscribe(topicsToSubscribe);
-            client.poll(1);
+            Map<String, List<PartitionInfo>> pi = client.listTopics();


Not final yet, will refactor it a bit as suggested in earlier comments.

viktorsomogyi · 2018-08-09T16:55:41Z

        try (final KafkaConsumer<byte[], byte[]> client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
-            client.subscribe(topicsToSubscribe);
-            client.poll(1);
+            Collection<TopicPartition> partitions = topicsToSubscribe.stream().map(client::partitionsFor)


@hachikuji were you thinking of something like this?

Java 8 makes this much nicer than I was expecting. 😄

viktorsomogyi · 2018-08-09T16:56:42Z

General update:

some fixes around suggestions from review
fixed test failures (some polls there needed to be replaced)
rebased on latest trunk

viktorsomogyi · 2018-08-09T17:08:27Z

retest this please

mjsax · 2018-08-09T18:48:45Z

@viktorsomogyi There is a checkstyle error

/home/jenkins/jenkins-slave/workspace/kafka-pr-jdk10-scala2.12/core/src/main/scala/kafka/tools/StreamsResetter.java:36:8: Unused import - org.apache.kafka.common.PartitionInfo. [UnusedImports

hachikuji

Thanks for the updates. Just one comment for discussion.

hachikuji · 2018-08-10T00:16:24Z

        try (final KafkaConsumer<byte[], byte[]> client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
-            client.subscribe(topicsToSubscribe);
-            client.poll(1);
+            Collection<TopicPartition> partitions = topicsToSubscribe.stream().map(client::partitionsFor)


Java 8 makes this much nicer than I was expecting. 😄

hachikuji · 2018-08-10T00:27:58Z

I guess another approach might be to produce a dummy record to the topic first before assigning the partitions to the consumer and starting the test. That would let us retain the existing behavior.

hachikuji · 2018-08-10T16:51:59Z

@viktorsomogyi Thanks for the update. LGTM overall. I think to fix the failing tests, we just need to use the right poll() in WorkerSinkTaskThreadedTest.

viktorsomogyi · 2018-08-10T17:28:30Z

Ah, indeed. If you haven't started fixing it yet, I pushed a commit with that. Let's see if we'll have any after this. :)

viktorsomogyi · 2018-08-10T20:22:30Z

It seems there was a flaky afterall. I'll rerun them.

viktorsomogyi · 2018-08-10T20:22:48Z

retest this please

tedyu · 2018-08-11T20:46:10Z

Can this be dropped from Consumer (in trunk) ?

    ConsumerRecords<K, V> poll(long timeout);

mjsax · 2018-08-12T02:20:21Z

@tedyu We cannot just remove a method from public API, as this would be a breaking change. Note, that the method is already deprecated though.

EndToEndLatency tool produces a dummy record in case the topic does not exist. This behavior was introduced in this PR #5319 as part of updating the tool to use latest consumer API. However, if we run the tool with producer acks == 1, the high watermark may not be updated before we reset consumer offsets to latest. In rare cases when this happens, the tool will throw an exception in the for loop where the consumer will unexpectedly consume the dummy record. As a result, we occasionally see Benchmark.test_end_to_end_latency system test failures. This PR checks if topic exists, and creates the topic using AdminClient if it does not exist. Author: Anna Povzner <anna@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io> Closes #5950 from apovzner/fix-EndToEndLatency (cherry picked from commit 3acebe6) Signed-off-by: Ewen Cheslack-Postava <me@ewencp.org>

EndToEndLatency tool produces a dummy record in case the topic does not exist. This behavior was introduced in this PR #5319 as part of updating the tool to use latest consumer API. However, if we run the tool with producer acks == 1, the high watermark may not be updated before we reset consumer offsets to latest. In rare cases when this happens, the tool will throw an exception in the for loop where the consumer will unexpectedly consume the dummy record. As a result, we occasionally see Benchmark.test_end_to_end_latency system test failures. This PR checks if topic exists, and creates the topic using AdminClient if it does not exist. Author: Anna Povzner <anna@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io> Closes #5950 from apovzner/fix-EndToEndLatency

EndToEndLatency tool produces a dummy record in case the topic does not exist. This behavior was introduced in this PR apache#5319 as part of updating the tool to use latest consumer API. However, if we run the tool with producer acks == 1, the high watermark may not be updated before we reset consumer offsets to latest. In rare cases when this happens, the tool will throw an exception in the for loop where the consumer will unexpectedly consume the dummy record. As a result, we occasionally see Benchmark.test_end_to_end_latency system test failures. This PR checks if topic exists, and creates the topic using AdminClient if it does not exist. Author: Anna Povzner <anna@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io> Closes apache#5950 from apovzner/fix-EndToEndLatency (cherry picked from commit 3acebe6) Signed-off-by: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Matthias J. Sax <mjsax@apache.org>, Jason Gustafson <jason@confluent.io>

EndToEndLatency tool produces a dummy record in case the topic does not exist. This behavior was introduced in this PR apache#5319 as part of updating the tool to use latest consumer API. However, if we run the tool with producer acks == 1, the high watermark may not be updated before we reset consumer offsets to latest. In rare cases when this happens, the tool will throw an exception in the for loop where the consumer will unexpectedly consume the dummy record. As a result, we occasionally see Benchmark.test_end_to_end_latency system test failures. This PR checks if topic exists, and creates the topic using AdminClient if it does not exist. Author: Anna Povzner <anna@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io> Closes apache#5950 from apovzner/fix-EndToEndLatency

viktorsomogyi commented Jul 9, 2018

View reviewed changes

viktorsomogyi changed the title ~~MINOR: use poll(Duration) in consumer example~~ KAFKA-7140: Remove deprecated poll usages Jul 9, 2018

kamalcph mentioned this pull request Aug 8, 2018

MINOR : kafka-examples, replaced deprecated KafkaConsumer.poll(long) method with KafkaConsumer.poll(Duration) #5476

Closed

hachikuji reviewed Aug 8, 2018

View reviewed changes

viktorsomogyi force-pushed the example_deprecated_poll branch 2 times, most recently from c53957e to 6ae1afa Compare August 9, 2018 14:58

viktorsomogyi commented Aug 9, 2018

View reviewed changes

viktorsomogyi added 7 commits August 9, 2018 18:43

MINOR: use poll(Duration) in consumer example

910317a

Replace other non-test usages as well

fe77b3c

Removing the rest of the deprecated non-test poll usages

ab31aea

Use empty iterator in ConsoleConsumer and manual assignment in EndToE…

a359c53

…ndLatency

Replace poll calls in WorkerSinkTaskTest

87a57b7

Replace polls in ErrorHandlingTaskTest

68c6197

Manual partition assignment in StreamsResetter

6ad6f5e

viktorsomogyi force-pushed the example_deprecated_poll branch from bcf8279 to 6ad6f5e Compare August 9, 2018 16:44

viktorsomogyi commented Aug 9, 2018

View reviewed changes

Use partitionsFor in StreamsResetter

3482e73

viktorsomogyi commented Aug 9, 2018

View reviewed changes

hachikuji reviewed Aug 10, 2018

View reviewed changes

viktorsomogyi added 2 commits August 10, 2018 08:13

Remove unused imports

608624e

Auto create topic in EndToEndLatency

131b422

Fix failing tests in WorkerSinkTaskThreadedTest

581292f

hachikuji merged commit 8a78d76 into apache:trunk Aug 11, 2018

viktorsomogyi deleted the example_deprecated_poll branch August 23, 2018 13:10

apovzner mentioned this pull request Nov 27, 2018

MINOR: Fix handling of dummy record in EndToEndLatency tool #5950

Closed

3 tasks

pengxiaolong pushed a commit to pengxiaolong/kafka that referenced this pull request Jun 14, 2019

KAFKA-7140; Remove deprecated poll usages (apache#5319)

d910958

Reviewers: Matthias J. Sax <mjsax@apache.org>, Jason Gustafson <jason@confluent.io>

Uh oh!

Conversation

viktorsomogyi commented Jul 2, 2018

Committer Checklist (excluded from commit message)

Uh oh!

ijuma commented Jul 2, 2018

Uh oh!

viktorsomogyi commented Jul 9, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viktorsomogyi commented Jul 9, 2018

Uh oh!

hachikuji commented Jul 19, 2018

Uh oh!

viktorsomogyi commented Aug 8, 2018

Uh oh!

viktorsomogyi commented Aug 8, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viktorsomogyi Aug 9, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viktorsomogyi commented Aug 9, 2018

Uh oh!

viktorsomogyi commented Aug 9, 2018

Uh oh!

mjsax commented Aug 9, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Aug 10, 2018

Uh oh!

viktorsomogyi commented Aug 10, 2018

Uh oh!

viktorsomogyi commented Aug 10, 2018

Uh oh!

viktorsomogyi commented Aug 10, 2018

Uh oh!

tedyu commented Aug 11, 2018

Uh oh!

mjsax commented Aug 12, 2018

Uh oh!

viktorsomogyi commented Aug 8, 2018 •

edited

Loading

viktorsomogyi Aug 9, 2018 •

edited

Loading